package streamforce.workload;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Properties;
import java.util.Random;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import streamforce.Stat;
import streamforce.crypto.message.PolicyMessage;
import streamforce.crypto.owner.EPOwnerClient;

/**
 * Generating policies of 5 different types:
 * 
 * T1: select * from StockEvent(stockId = x) - 1
 * T2: select * from StockEvent(stockId = x, y < timestamp < z) - 20
 * T3: select * from StockEvent(stockId = x, y < hour < z) - 11
 * T4: select * from StockEvent(stockId = x, timestamp % y = z) - 20
 * T5: select sum(price), sum(volume) from StockEvent(stockId=x).win:length_batch(y)  - 8 
 * T6: join price (with other conditions) - 40 (10 from T2, 10 from T4) 
 */
public class PolicyGenerator {
	static Random rand;
	static Log log = LogFactory.getLog(PolicyGenerator.class);
	static ArrayList<PolicyMessage> policyMessages = new ArrayList<PolicyMessage>(); 
	
	//<streamId> <userBaseId>  
	public static void main(String[] args) throws Exception{
		int streamId = new Integer(args[0]).intValue();		
		int userBaseId = new Integer(args[1]).intValue(); 
		
		rand = new Random(streamId); 
		
		Properties prop = new Properties();
		prop.load(new FileInputStream("attributes.properties"));					
		String[] ws = prop.getProperty("windowSizes").split(",");
		int[] windowSizes = new int[ws.length];
		for (int i=0; i<ws.length; i++){
			windowSizes[i] = new Integer(ws[i]).intValue();
		}
		
		String policyFile = prop.getProperty("policyFile"); 
		ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(policyFile+"_"+streamId));
		
		
		EPOwnerClient owner = new EPOwnerClient(streamId);			
		generateType1(owner, streamId, userBaseId); 
		userBaseId+=1; 
		generateType2(owner, 20, streamId, userBaseId); 
		userBaseId+=20;
		generateType3(owner, 11, streamId, userBaseId); 
		userBaseId+=11; 
		generateType4(owner, 20, streamId, userBaseId,windowSizes); 
		userBaseId+=20; 
		generateType5(owner, streamId, userBaseId, windowSizes); 
		userBaseId+=8; 
		generateType6(owner,40,streamId,userBaseId,windowSizes); 
		oos.writeObject(policyMessages);
		oos.flush(); 
		oos.close(); 
	}
	
	//type 1 - 1
	public static void generateType1(EPOwnerClient owner, int streamId, int userIdBase){
		PolicyInfo[] s = new PolicyInfo[1];
		s[0] = new PolicyInfo(streamId,userIdBase,"stockId_"+streamId, "stockId="+streamId);
		long start = System.currentTimeMillis(); 
		PolicyMessage pm = EPOwnerClient.initSimpleClient(owner, streamId, s[0]); 
		PolicyGenerator.policyMessages.add(pm); 
		long end = System.currentTimeMillis(); 		
		log.info("policy init - Type 1 = "+(end-start));
		log.info("policy "+userIdBase+": "+s[0].getPolicy());
		
	}
	
	//type 2 - 20
	public static void generateType2(EPOwnerClient owner, int n, int streamId, int userIdBase){
		PolicyInfo[] s = new PolicyInfo[n];
		//at least for 2 days
		PolicyMessage pm; 
		for (int i = 0; i<n; i++){
			int y = rand.nextInt(300)+1;
			int z = rand.nextInt(7000)+3000;
			s[i] = new PolicyInfo(streamId, userIdBase + i, "stockId_"
					+ streamId + " and timestamp > " + y + " and timestamp < "
					+ z, "stockId="+streamId+", timestamp > "+y+", timestamp < "+z);
			long start = System.currentTimeMillis(); 
			pm = EPOwnerClient.initSimpleClient(owner, streamId, s[i]);
			policyMessages.add(pm); 
			long end = System.currentTimeMillis(); 
			log.info("policy init - Type 2 = "+(end-start)); 
			log.info("policy "+s[i].getUserId()+": "+s[i].getPolicy());
		}				
	}
	
	//type 3 - 11
	public static void generateType3(EPOwnerClient owner, int n, int streamId, int userIdBase){
		PolicyInfo[] s = new PolicyInfo[n];
		int startH = 1;
		int endH = 13; 
		PolicyMessage pm; 
		for (int i=0; i<n; i++){			
			startH = i+1; 
			endH = Math.min(startH + rand.nextInt(12),23); 
			s[i] = new PolicyInfo(streamId, userIdBase + i, "stockId_"
					+ streamId + " and hour >= " + startH + " and hour <= "
					+ endH, "stockId=" + streamId + ", realTime >=" + startH
					+ ", realTime <=" + endH);
			long start = System.currentTimeMillis(); 
			pm = EPOwnerClient.initSimpleClient(owner, streamId, s[i]); 
			policyMessages.add(pm); 
			long end = System.currentTimeMillis(); 
			log.info("policy init - Type 3 = "+(end-start)); 
			log.info("policy "+s[i].getUserId()+": "+s[i].getPolicy());
		}		
	}
		
	//type 4 - 20
	public static void generateType4(EPOwnerClient owner, int n, int streamId, int userIdBase, int[] windowSizes){
		PolicyInfo[] s = new PolicyInfo[n];
		PolicyMessage pm; 
		for (int i=0; i<n; i++){
			int j = rand.nextInt(windowSizes.length);	
			int ws = rand.nextInt(windowSizes[j]); 
			s[i] = new PolicyInfo(streamId, userIdBase + i, "stockId_"
					+ streamId + " and timestamp % " + windowSizes[j] + " = "
					+ ws, "stockId=" + streamId + ", timestamp % "
					+ windowSizes[j] + "=" + ws);		
			long start = System.currentTimeMillis(); 
			pm = EPOwnerClient.initSimpleClient(owner, streamId, s[i]);
			policyMessages.add(pm);
			long end = System.currentTimeMillis(); 
			log.info("policy init - Type 4 = "+(end-start)); 
			log.info("policy "+s[i].getUserId()+": "+s[i].getPolicy());
		}		
	}
	
	//type 5 - 8
	public static void generateType5(EPOwnerClient owner, int streamId, int userIdBase, int[] windowSizes){
		PolicyInfo[] s = new PolicyInfo[windowSizes.length];
		PolicyMessage pm;
		int ws; 
		for (int i = 0; i < s.length; i++) {
			ws = windowSizes[rand.nextInt(windowSizes.length)];
			s[i] = new PolicyInfo(streamId, userIdBase + i, "stockId_"
					+ streamId + " and ws_"
					+ ws, "stockId="+streamId, ws);
			long start = System.currentTimeMillis(); 			
			pm = EPOwnerClient.initAggregateClient_2(owner, streamId, s[i]); 
			policyMessages.add(pm); 
			long end = System.currentTimeMillis(); 
			log.info("policy init - Type 5 = "+(end-start));
			log.info("policy "+s[i].getUserId()+": "+s[i].getPolicy());
		}		
		
		
	}
	
	//type 6 - 40 (join price)
	public static void generateType6(EPOwnerClient owner, int n, int streamId, int userIdBase, int[] windowSizes){
		PolicyInfo[] s = new PolicyInfo[n];
		//at least for 2 days
		PolicyMessage pm; 
		for (int i = 0; i<n/2; i++){
			int y = rand.nextInt(300)+1;
			int z = rand.nextInt(7000)+3000;
			s[i] = new PolicyInfo(streamId, userIdBase + i, "stockId_"
					+ streamId + " and timestamp > " + y + " and timestamp < "
					+ z + " and att_volume", "stockId="
					+ streamId + ", timestamp > " + y + ", timestamp < " + z);
			long start = System.currentTimeMillis(); 
			pm = EPOwnerClient.initJoinClient(owner, streamId, s[i]);
			policyMessages.add(pm); 
			long end = System.currentTimeMillis(); 
			log.info("policy init - Type 6 = "+(end-start)); 
			log.info("policy join "+s[i].getUserId()+": "+s[i].getPolicy());
	
		}		
		
		for (int i=0; i<n/2; i++){
			int j = rand.nextInt(windowSizes.length);	
			int ws = rand.nextInt(windowSizes[j]); 
			s[i] = new PolicyInfo(streamId, userIdBase + i+n/2, "stockId_"
					+ streamId + " and timestamp % " + windowSizes[j] + " = "
					+ ws + " and att_volume", "stockId="
					+ streamId + ", timestamp % " + windowSizes[j] + "=" + ws);		
			long start = System.currentTimeMillis(); 
			pm = EPOwnerClient.initJoinClient(owner, streamId, s[i]);
			policyMessages.add(pm);
			long end = System.currentTimeMillis(); 
			log.info("policy init - Type 6 = "+(end-start)); 
			log.info("policy join "+s[i].getUserId()+": "+s[i].getPolicy());
		}				
	}		
}
